package org.example.using;


import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;

import static org.apache.spark.sql.functions.col;
import static org.apache.spark.sql.functions.expr;

import org.apache.spark.sql.SparkSession;
import org.example.bean.*;
import org.example.utils.MysqlUtils_2;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;


/***
 * 创建一个类,用来测试 spark 2.3.0版本的 DataFrame的用法, DataSet API的使用.
 */
public class Spark01_SparkDataSetSelectUsing {
    public static void main(String[] args) throws IOException {
        // 初始化SparkSession
        // 引入log4j,消除控制台的警告
        Logger.getLogger("org").setLevel(Level.WARN);
        Logger.getLogger("akka").setLevel(Level.WARN);
        SparkConf conf = new SparkConf().setAppName("PantographModelFunction") // 设置应用程序名称
                .setMaster("local[*]") // 设置运行模式，这里是本地模式
                .set("spark.executor.memory", "6g") // 设置executor内存大小
                .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") // 设置序列化器
                .set("spark.driver.allowMultipleContexts", "true") // 允许多个SparkContext同时存在
                .set("spark.hadoop.validateOutputSpecs", "false") // 禁止输出路径已经存在时抛出异常
                .set("hive.mapred.supports.subdirectories", "true") // 开启Hive支持
                .set("mapreduce.input.fileinputformat.input.dir.recursive", "true") // 开启递归读取文件夹中的文件
                .set("spark.debug.maxToStringFields", "100").set("spark.sql.codegen.wholeStage", "false")
                .set("spark.sql.caseSensitive", "false"); // Spark会自动处理大小写不匹配的情况
        JavaSparkContext sc = new JavaSparkContext(conf);
        SparkSession spark = SparkSession.builder().config(conf).getOrCreate();

        /*
            TODO 读取MySQL中的表,根据表结构创建DF.这里需要提前手动创建一个 Java Bean,这里使用MySQL中的一张表进行演示.表中的字段全部转化为小写.
                 在使用过程中, 需要注意MySQL表定义时的字段类型,也就是需要注意时间格式. 关于时间格式的转化在`MysqlUtils_2`工具类中读取表时,手动添加了如下配置项:
                    .option("dateFormat", "yyyy-MM-dd") // 根据需要设置日期格式
                    .option("timestampFormat", "yyyy-MM-dd HH:mm:ss") // 根据需要设置时间戳格式
                 关于Java Bean 的定义上:
                    a. 在定义Java Bean时，确保为每个属性提供getter和setter方法，以便Spark正确地访问和设置属性值。
                    b. 您的Java Bean应该实现Serializable接口，以便在Spark集群中进行分布式处理。
                    c. 请确保使用与数据库驱动程序和MySQL版本兼容的Spark连接器。
                 关于MySQL表字段的类型,如果是时间格式,例如: s_htsj、s_hksj和s_xfsj 列的类型是 datetime，所以在Java Bean中使用java.sql.Timestamp类型.不要导错包
         */
        Dataset<AdsPhmFault> adsPhmFaultDF = MysqlUtils_2.readMySQLTableDF(spark, "ads_phm_fault", AdsPhmFault.class);
        adsPhmFaultDF.show(false);

        Dataset<TPhmGzzdYjgzbjRank> tPhmGzzdYjgzbjRankDF = MysqlUtils_2.readMySQLTableDF(spark, "t_phm_gzzd_yjgzbj_rank", TPhmGzzdYjgzbjRank.class);
        tPhmGzzdYjgzbjRankDF.show(false);

        Dataset<TPhmMxxq> tPhmMxxq = MysqlUtils_2.readMySQLTableDF(spark, "t_phm_mxxq", TPhmMxxq.class);
        tPhmMxxq.show(false);

        Dataset<TPhmSqwx> tPhmSqwxDF = MysqlUtils_2.readMySQLTableDF(spark, "t_phm_sqwx", TPhmSqwx.class);
        tPhmSqwxDF.show(false);
        /**
         * TODO 选择指定的列,并返回查询结果.
         *      可以直接使用 下面的三种写法
         *      - 第一种直接使用: select("","","" ...) 字段名称,返回一个 DataSet<Row> 类型
         *      - 第二种写法使用: tPhmSqwxDF.select(tPhmSqwxDF.col("S_TRAIN_ID"),...),返回的结果也是一个 DataSet<Row> 类型
         *      - 第三种写法使用: selectExpr ,这种写法允许你在" " 内写sql表达式,使用起开也很好用,使用sparksql中的一些函数,可以直接写就好.eg: rowDataset 这里
         *      - 第四种写法使用: expr ,这种写法允许你在" " 内写sql表达式,使用起开也很好用,使用sparksql中的一些函数,可以直接写就好.eg: rowDataset1 这里
         *      - 第五种,需要返回指定类型的DataSet.也就是指定实体类的DS.需要将返回的结果进行类型转换,写法看下面的第五种写法.
         *
         */
        // 第一种写法
        Dataset<Row> selectedDF = tPhmSqwxDF.select("S_TRAIN_ID", "MXMC", "CREATE_TIME", "CZJY");
        selectedDF.show(false);
        // 第二种写法
        Dataset<Row> selectedDF_2 = tPhmSqwxDF.select(col("S_TRAIN_ID"), col("MXMC"), tPhmSqwxDF.col("CZJY"));
        selectedDF_2.show(false);
        // 第三种写法使用: 获取到当前数据集中的某几个字段的组合,并返回一个新的列用来标识它的唯一. 组合规则: S_TRAIN_ID + BJWZ + CREATE_TIME,组合结束后可以起一个别名.
        Dataset<Row> rowDataset = tPhmSqwxDF.selectExpr("concat_ws('-',S_TRAIN_ID,BJWZ,CREATE_TIME)  as idx", "XTPJM", "BJPJM");
        rowDataset.show(false);
        /*
            这里在使用的时候需要稍微注意一下:spark sql我这里使用的是 2.3.0版本, `expr`是 org.apache.spark.sql.functions下的方法,有些时候导入的时候直接写会爆红,可能是我的IDEA的问题,但是这个肯定有的.如果爆红,就直接 import org.apache.spark.sql.functions.col;这里
            +----------+-------------------------------------------+------------------------------------------------------------------------+
            |S_TRAIN_ID|s_train_mc                                 |MXMS                                                                    |
            +----------+-------------------------------------------+------------------------------------------------------------------------+
            |HXD1D-1001|HXD1D-1001-元件故障                            |斩波管故障                                                             |
         */
        // 第四种写法
        Dataset<Row> rowDataset1 = tPhmSqwxDF.select(col("S_TRAIN_ID"), expr("concat_ws('-',S_TRAIN_ID,MXMC) as s_train_mc"), col("MXMS"));
        rowDataset1.show(false);

        // 第五种写法,返回指定类型的DataSet.
        String[] jsonData = new String[]{
                "[" +
                        "{\"s_train_id\": \"HXD1D-1066\",\"CX\": \"HXD1D\",\"phm_mc\": \"变流器1L1元件总故障\",\"phm_lb\": \"变流器元件故障报警\",\"GZKSSJ_UP\": \"2023-03-17 10:00:14\",\"phm_bjwz\": \"1\",\"LJ\": \"路局\",\"ljjc\": \"路局简称\",\"LJ_ID\": \"路局码\",\"PSD\": \"配属段\",\"psdjc\": \"配属段简称\",\"PSD_ID\": \"配属段代码\",\"phm_xt\": \"所属系统\",\"phm_xtbm\": \"系统编码\",\"phm_xtpjm\": \"系统PJM\",\"phm_bj\": \"所属部件\",\"phm_bjbm\": \"故障部位码1\",\"phm_pjm\": \"部件PJM\",\"pjxlh\": \"配件序列号\",\"pjdjm\": \"配件单件码\",\"IDX\": \"1\",\"SOURCE\": \"故障源\",\"GZDM_WM\": \"故障码\",\"GZMC\": \"故障名称\",\"zj_ms\": \"故障描述-株机\",\"zj_fa\": \"施修方法-株机\",\"cksj\": \"2022-01-02 12:30:55\"}" +
                        "]"
        };
        // 将当前消费到的 json类型的字符串,转化成为 RDD,在将当前类型给的RDD转化为指定类型的 DataSet,这里将json字符串中的属性,封装成一个实体类:KafkaJsonData
        JavaRDD<String> jsonRDD = sc.parallelize(Arrays.asList(jsonData));
        Dataset<Row> jsonDs = spark.read().json(jsonRDD);
        processFaultType1(spark, jsonDs).get("kafkaJsonData").show(false);


        spark.stop();
    }

    /***
     * 定义一个方法,传入的参数 inputData 是消费kafka中实时数据后的数据以及处理其他数据,例如:视情维修处理结果是一个参数,健康评估是一个参数,最终写入到`ads_phm_fault`的结果也是一个参数.最终的几个参数做集中处理.
     * @param spark
     * @param inputData
     * @throws IOException
     */
    public static Map<String, Dataset<?>> processFaultType1(SparkSession spark, Dataset<Row> inputData) throws IOException {
        Dataset<KafkaJsonData> kafkaJsonDataDataset = inputData.as(Encoders.bean(KafkaJsonData.class));
        Dataset<TPhmSqwx> tPhmSqwxDF = MysqlUtils_2.readMySQLTableDF(spark, "t_phm_sqwx", TPhmSqwx.class);
        Dataset<TPhmGzzdYjgzbjRank> tPhmGzzdYjgzbjRankDF = MysqlUtils_2.readMySQLTableDF(spark, "t_phm_gzzd_yjgzbj_rank", TPhmGzzdYjgzbjRank.class);
        Map<String, Dataset<?>> resultMap = new HashMap<>();
        resultMap.put("kafkaJsonData", kafkaJsonDataDataset);
        resultMap.put("tPhmSqwx", tPhmSqwxDF);
        resultMap.put("tPhmGzzdYjgzbjRank", tPhmGzzdYjgzbjRankDF);
        return resultMap;
    }
}
